package com.hubiao.pro.service.consumer;

import com.alibaba.fastjson.JSON;
import com.hubiao.pro.common.constant.Constants;
import com.hubiao.pro.common.redis.Cache;
import com.hubiao.pro.common.utils.StringUtils;
import com.hubiao.pro.domain.db.User;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * Description  队列消费者
 *
 * @author hubiao
 * @since 2021-01-30 22:20
 */
@Slf4j
@Component
public class ConsumerDealInfo {


    /*** 可用处理器的Java虚拟机的CPU数量 */
    private static final int availableProcessor = Runtime.getRuntime().availableProcessors();
    /*** 核心线程数 */
    int corePoolSize = availableProcessor * 2 + 1;
    /*** 最大线程数 */
    int maxPoolSize = (availableProcessor * 2 + 1) * 2;
    ThreadPoolExecutor service = new ThreadPoolExecutor( corePoolSize, maxPoolSize,
            15L, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>( 1024 ),
            new ThreadPoolExecutor.AbortPolicy() );
    @Resource
    private Cache cache;

    /**
     * 测试消息对接的解决方案
     *
     * @param message
     */
    @RabbitListener(queues = Constants.MONITOR_QUEUE_TEST)
    public void userSend(Message message) {
        // 幂等处理
        String key = StringUtils.format( Constants.MQ_KEY, message.getMessageProperties().getMessageId() );
        if (!cache.exists( key )) {
            return;
        }

        service.execute( () -> dealInf( message ) );
    }

    public void dealInf(Message message) {
        User user = JSON.parseObject( new String( message.getBody() ), User.class );
        System.out.println( "user = " + user );
    }

}
